package com.bestvike.stone.spring.redis.lock;

import com.bestvike.linq.Linq;
import com.bestvike.linq.exception.ArgumentException;
import com.bestvike.stone.core.lang.StringUtils;
import com.bestvike.stone.spring.concurrent.IDistributedLock;
import com.bestvike.stone.spring.concurrent.LockFailedException;
import com.bestvike.stone.spring.context.TraceContext;
import com.bestvike.stone.spring.redis.RedisProperties;
import com.bestvike.stone.spring.redis.RedisUtils;
import com.bestvike.stone.spring.redis.converter.LongRedisSerializer;
import com.bestvike.stone.spring.redis.converter.ObjectRedisSerializer;
import lombok.extern.apachecommons.CommonsLog;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;

import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.TimeUnit;

/**
 * Redis 可重入分布式锁
 * Created by 许崇雷 on 2017-11-15.
 */
@CommonsLog
public final class RedisDistributedLock implements IDistributedLock<RedisLockObject> {
    /**
     * 构造函数
     */
    public RedisDistributedLock() {
    }

    /**
     * 根据当前线程和追踪上下文获取锁对象标识
     */
    private static String getLockId() {
        if (TraceContext.exists())
            return Thread.currentThread().getId() + "@" + TraceContext.getTraceId();
        throw new RuntimeException("no trace context, can not get lock id.");
    }

    /**
     * 根据资源名称获取锁对象名称
     */
    private static String getLockName(String name) {
        return Config.NAME_PREFIX + name + Config.NAME_SUFFIX;
    }

    /**
     * 根据资源名称获取订阅频道名称
     */
    private static String getChannelName(String name) {
        return Config.CHANNEL_PREFIX + name + Config.CHANNEL_SUFFIX;
    }

    /**
     * 申请锁
     *
     * @param name      资源名称
     * @param waitTime  等待时间
     * @param leaseTime 持有时间
     * @param unit      时间单位
     * @return 成功返回锁对象, 否则抛出异常
     */
    @Override
    public RedisLockObject acquire(String name, long waitTime, long leaseTime, TimeUnit unit) throws LockFailedException {
        if (StringUtils.isEmpty(name))
            throw new ArgumentException("name can not be empty.");
        if (leaseTime <= 0)
            throw new ArgumentException("leaseTime must greater than zero.");
        if (unit == null)
            throw new ArgumentException("unit can not be null.");
        if (!Linq.asEnumerable(name).all(c -> c >= '0' && c <= '9' || c >= 'A' && c <= 'Z' || c >= 'a' && c <= 'z' || c == '-' || c == ':' || c == '_'))
            throw new ArgumentException("name can not contains special characters.");

        final long waitEnd = System.currentTimeMillis() + unit.toMillis(waitTime);//wait end timestamp
        final String lockId = getLockId();//field
        final String lockName = getLockName(name);//key
        final String channelName = getChannelName(name);////pub sub topic name
        final long leaseMillis = unit.toMillis(leaseTime);//lease millis
        final RedisLockObject lockObject = new RedisLockObject(this, lockId, lockName, channelName, leaseMillis);//lock object
        final MessageListener messageListener = new RedisLockMessageListener(lockObject);//pub sub listener
        final Topic pubSubTopic = new ChannelTopic(Config.REDIS_PROPERTIES.addPrefix(channelName));//pub sub topic
        do {
            //申请锁
            if (RedisUtils.execute(AcquireConfig.REDIS_SCRIPT, ObjectRedisSerializer.DEFAULT, LongRedisSerializer.DEFAULT, Collections.singletonList(lockName), lockId, leaseMillis) == null) {
                log.info(String.format(AcquireConfig.SUCCESS_LOG, Config.REDIS_PROPERTIES.addPrefix(lockName), lockId));
                return lockObject;
            }

            //等待截止
            if (System.currentTimeMillis() >= waitEnd) {
                log.info(String.format(AcquireConfig.FAILED_LOG, Config.REDIS_PROPERTIES.addPrefix(lockName), lockId));
                throw new LockFailedException(String.format(AcquireConfig.FAILED_MSG, Config.REDIS_PROPERTIES.addPrefix(lockName), lockId));
            }

            //noinspection SynchronizationOnLocalVariableOrMethodParameter
            synchronized (lockObject) {
                try {
                    //订阅
                    Config.MESSAGE_LISTENER_CONTAINER.addMessageListener(messageListener, pubSubTopic);

                    //等待
                    lockObject.wait(Math.max(1, waitEnd - System.currentTimeMillis()));

                    //等待截止
                    if (System.currentTimeMillis() >= waitEnd) {
                        log.info(String.format(AcquireConfig.FAILED_LOG, Config.REDIS_PROPERTIES.addPrefix(lockName), lockId));
                        throw new LockFailedException(String.format(AcquireConfig.FAILED_MSG, Config.REDIS_PROPERTIES.addPrefix(lockName), lockId));
                    }
                } catch (InterruptedException e) {
                    //线程中断
                    log.info(String.format(AcquireConfig.FAILED_LOG, Config.REDIS_PROPERTIES.addPrefix(lockName), lockId));
                    throw new LockFailedException(String.format(AcquireConfig.FAILED_MSG, Config.REDIS_PROPERTIES.addPrefix(lockName), lockId), e);
                } finally {
                    //取消订阅
                    Config.MESSAGE_LISTENER_CONTAINER.removeMessageListener(messageListener, pubSubTopic);
                }
            }
        } while (true);
    }

    /**
     * 续订锁
     *
     * @param lockObject 锁对象
     */
    @Override
    public void renewal(RedisLockObject lockObject) {
        if (lockObject == null)
            throw new ArgumentException("lockObject can not be null.");

        final String lockId = lockObject.getLockId();
        final String lockName = lockObject.getLockName();
        final long leaseMillis = lockObject.getLeaseMillis();
        RedisUtils.execute(RenewalConfig.REDIS_SCRIPT, ObjectRedisSerializer.DEFAULT, LongRedisSerializer.DEFAULT, Collections.singletonList(lockName), lockId, leaseMillis);
        log.info(String.format(RenewalConfig.LOG, Config.REDIS_PROPERTIES.addPrefix(lockName), lockId));
    }

    /**
     * 释放锁
     *
     * @param lockObject 锁对象
     */
    @Override
    public void release(RedisLockObject lockObject) {
        if (lockObject == null)
            throw new ArgumentException("lockObject can not be null.");

        final String lockId = lockObject.getLockId();
        final String lockName = lockObject.getLockName();
        final String channelName = lockObject.getChannelName();
        final long leaseMillis = lockObject.getLeaseMillis();
        RedisUtils.execute(ReleaseConfig.REDIS_SCRIPT, ObjectRedisSerializer.DEFAULT, LongRedisSerializer.DEFAULT, Arrays.asList(lockName, channelName), lockId, leaseMillis, Config.RELEASE_MESSAGE);
        log.info(String.format(ReleaseConfig.LOG, Config.REDIS_PROPERTIES.addPrefix(lockName), lockId));
    }


    private static final class Config {
        private static final String NAME_PREFIX = "LOCK:{";//锁对象名称前缀
        private static final String NAME_SUFFIX = "}";//锁对象名称后缀
        private static final String CHANNEL_PREFIX = "LOCK_CHANNEL:{";//订阅频道名称前缀
        private static final String CHANNEL_SUFFIX = "}";//订阅频道名称后缀
        private static final Long RELEASE_MESSAGE = 0L;//解锁消息
        private static final RedisProperties REDIS_PROPERTIES = RedisUtils.getRedisProperties();//配置
        private static final RedisMessageListenerContainer MESSAGE_LISTENER_CONTAINER = RedisUtils.getRedisMessageListenerContainer();//监听器容器
    }

    private static final class AcquireConfig {
        private static final String SUCCESS_LOG = "==>Redis Acquire Lock: %s %s";
        private static final String FAILED_LOG = "==>Redis Acquire Lock Failed: %s %s";
        private static final String FAILED_MSG = "Redis Acquire Lock Failed: %s %s";
        private static final String SCRIPT = "if (redis.call('exists', KEYS[1]) == 0) then " +
                "redis.call('hset', KEYS[1], ARGV[1], 1); " +
                "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                "return nil; " +
                "end; " +
                "if (redis.call('hexists', KEYS[1], ARGV[1]) == 1) then " +
                "redis.call('hincrby', KEYS[1], ARGV[1], 1); " +
                "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                "return nil; " +
                "end; " +
                "return redis.call('pttl', KEYS[1]);";
        private static final RedisScript<Long> REDIS_SCRIPT = new DefaultRedisScript<>(SCRIPT, Long.class);
    }

    private static final class RenewalConfig {
        private static final String LOG = "==>Redis Renewal Lock: %s %s";
        private static final String SCRIPT = "if (redis.call('hexists', KEYS[1], ARGV[1]) == 1) then " +
                "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                "return 1; " +
                "end; " +
                "return 0;";
        private static final RedisScript<Long> REDIS_SCRIPT = new DefaultRedisScript<>(SCRIPT, Long.class);
    }

    private static final class ReleaseConfig {
        private static final String LOG = "==>Redis Release Lock: %s %s";
        private static final String SCRIPT = "if (redis.call('exists', KEYS[1]) == 0) then " +
                "redis.call('publish', KEYS[2], ARGV[3]); " +
                "return 1; " +
                "end; " +
                "if (redis.call('hexists', KEYS[1], ARGV[1]) == 0) then " +
                "return nil; " +
                "end; " +
                "local counter = redis.call('hincrby', KEYS[1], ARGV[1], -1); " +
                "if (counter > 0) then " +
                "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                "return 0; " +
                "end; " +
                "redis.call('del', KEYS[1]); " +
                "redis.call('publish', KEYS[2], ARGV[3]); " +
                "return 1;";
        private static final RedisScript<Long> REDIS_SCRIPT = new DefaultRedisScript<>(SCRIPT, Long.class);
    }

    private static final class RedisLockMessageListener implements MessageListener {
        private final Object lockObject;

        RedisLockMessageListener(Object lockObject) {
            this.lockObject = lockObject;
        }

        @Override
        public void onMessage(Message message, byte[] pattern) {
            synchronized (this.lockObject) {
                this.lockObject.notify();
            }
        }
    }
}
